package com.zy.demo;

import com.zy.constant.MqConstant;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.junit.jupiter.api.Test;

import java.util.LinkedList;
import java.util.List;

/**
 * 顺序消息，保证入队顺序一致，进入一个主题中同一个队列
 */
public class OrderlyTest {

    @Test
    public void testOrderly() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("orderly-produce-group");
        producer.setNamesrvAddr(MqConstant.NAMESRV_ADDR);
        producer.start();
        List<Student> list = new LinkedList<>();
        list.add(new Student(10, "张三"));
        list.add(new Student(10, "李四"));
        list.add(new Student(10, "王五"));
        list.add(new Student(11, "幅度萨芬"));
        list.add(new Student(11, "放大"));
        list.add(new Student(11, "范德萨"));
        list.forEach(student -> {
            try {
                producer.send(new Message("orderly-topic",student.getName().getBytes()),new MessageQueueSelector() {

                    @Override
                    public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                        int grade = (int) o;
                        int i = grade % list.size();
                        return list.get(i);
                    }
                },student.getGrade());
            } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
                throw new RuntimeException(e);
            }
        });

        producer.shutdown();

    }

    @Test
    public void testConsumer() throws Exception {

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("orderly-consumer-group");

        consumer.setNamesrvAddr(MqConstant.NAMESRV_ADDR);
        consumer.subscribe("orderly-topic", "*");
        //每个队列单线程消费
        consumer.registerMessageListener(new MessageListenerOrderly(){

            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                System.out.println(Thread.currentThread().getName() + " consume message :" + new String(list.get(0).getBody()));
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });

        consumer.start();
        System.in.read();

    }




}

@Data
@AllArgsConstructor
@NoArgsConstructor
class Student{
    private Integer grade;
    private String name;
}